package com.ruoyi.security.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.common.collect.Lists;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.security.constant.ApiConstant;
import com.ruoyi.security.convert.SecurityDayMinuteTradeConvert;
import com.ruoyi.security.domain.SecurityDayMinuteTrade;
import com.ruoyi.security.domain.SecurityTradingCalendar;
import com.ruoyi.security.dto.SecurityDayMinuteTradeDto;
import com.ruoyi.security.dto.SyncSecurityDayMinuteTradeDto;
import com.ruoyi.security.mapper.SecurityDayMinuteTradeMapper;
import com.ruoyi.security.response.SecurityDayMinuteTradeResponse;
import com.ruoyi.security.service.ISecurityDayMinuteTradeService;
import com.ruoyi.security.service.ISecurityTradingCalendarService;
import com.ruoyi.security.service.craw.ICrawSecurityDayMinuteTradeService;
import com.ruoyi.security.util.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @author yeoman
 * @date 2025/4/30 17:12
 * @description
 */
@Slf4j
@Service
public class SecurityDayMinuteTradeServiceImpl implements ISecurityDayMinuteTradeService {

    @Resource
    private SecurityDayMinuteTradeMapper securityDayMinuteTradeMapper;

    @Resource
    private ICrawSecurityDayMinuteTradeService crawService;
    
    @Resource
    private SyncTaskManager syncTaskManager;
    
    @Resource
    private ISecurityTradingCalendarService securityTradingCalendarService;
    
    /**
     * 线程池，用于并发同步分时数据
     * 核心线程数：5，最大线程数：5，队列大小：100
     */
    private final ExecutorService executorService = new ThreadPoolExecutor(
            5, 5, 60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            r -> {
                Thread thread = new Thread(r, "sync-timeshare-thread-" + System.currentTimeMillis());
                thread.setDaemon(true);
                return thread;
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    @Override
    public List<SecurityDayMinuteTrade> list(SecurityDayMinuteTradeDto dto) {
        LambdaQueryWrapper<SecurityDayMinuteTrade> queryWrapper = buildQueryConditions(dto);
        return securityDayMinuteTradeMapper.selectList(queryWrapper);
    }

    @Override
    public void syncList(SyncSecurityDayMinuteTradeDto dto) {
        // 获取需要同步的日期列表
        List<String> syncDateList = getSyncDateList(dto);
        if (CollectionUtils.isEmpty(syncDateList)) {
            return;
        }
        
        // 使用多线程并发同步数据（不使用任务管理，保持原有逻辑）
        syncListWithMultiThread(dto, syncDateList, null);
    }
    
    /**
     * 异步同步分时数据（返回任务ID）
     * 
     * @param dto 同步参数
     * @return 任务ID
     */
    @Override
    public String syncListAsync(SyncSecurityDayMinuteTradeDto dto) {
        // 获取需要同步的日期列表
        List<String> syncDateList = getSyncDateList(dto);
        if (CollectionUtils.isEmpty(syncDateList)) {
            return null;
        }
        
        // 创建任务
        String taskId = syncTaskManager.createTask(dto.getSymbol(), syncDateList.size());
        
        // 直接提交所有日期任务到线程池，不使用外层异步包装
        // 这样可以充分利用线程池的并发能力（核心线程数2，最大线程数5）
        // 所有日期任务会并发执行，最多5个任务同时运行
        for (String date : syncDateList) {
            CompletableFuture.runAsync(() -> {
                try {
                    log.info("开始同步分时数据，证券代码：{}，日期：{}", dto.getSymbol(), date);
                    
                    // 更新任务进度
                    int currentSynced = syncTaskManager.getTask(taskId).getSyncedDays().get();
                    syncTaskManager.updateProgress(taskId, currentSynced, date);
                    
                    // 调用爬虫服务获取单个日期的数据
                    List<String> dateList = Lists.newArrayList(date);
                    List<SecurityDayMinuteTradeResponse> responseList = crawService.crawList(dto.getSymbol(), dateList);
                    
                    if (CollectionUtils.isEmpty(responseList)) {
                        log.warn("证券【{} {}】未获取到数据", dto.getSymbol(), date);
                        // 未获取到数据，但日期已处理，计入已处理天数
                        syncTaskManager.incrementProcessedDays(taskId);
                        return;
                    }
                    
                    // 转换为实体对象
                    List<SecurityDayMinuteTrade> insertList = SecurityDayMinuteTradeConvert.INSTANCE.toSecurityDayMinuteTradeList(responseList);
                    
                    // 设置字段
                    Date now = new Date();
                    for (SecurityDayMinuteTrade trade : insertList) {
                        trade.setSymbol(dto.getSymbol());
                        trade.setCode(dto.getSymbol().substring(2));
                        trade.setCreateTime(now);
                        trade.setUpdateTime(now);
                    }
                    
                    // 批量插入数据
                    if (!CollectionUtils.isEmpty(insertList)) {
                        for (SecurityDayMinuteTrade trade : insertList) {
                            securityDayMinuteTradeMapper.insert(trade);
                        }
                        log.info("同步分时数据成功，证券代码：{}，日期：{}，数据条数：{}", 
                                dto.getSymbol(), date, insertList.size());
                        
                        // 更新任务进度（成功同步）
                        syncTaskManager.incrementSyncedDays(taskId);
                        syncTaskManager.incrementProcessedDays(taskId);
                    } else {
                        // insertList为空，但已经处理过这个日期了
                        syncTaskManager.incrementProcessedDays(taskId);
                    }
                } catch (Exception e) {
                    log.error("同步分时数据失败，证券代码：{}，日期：{}", dto.getSymbol(), date, e);
                    // 同步失败，但日期已处理，计入失败和已处理
                    syncTaskManager.incrementFailedDays(taskId);
                    syncTaskManager.incrementProcessedDays(taskId);
                }
            }, executorService);
        }
        
        // 注意：不需要等待所有任务完成，因为每个任务完成时会自动调用 incrementProcessedDays
        // incrementProcessedDays 会自动检查任务是否完成并更新状态
        // 这样可以充分利用线程池的并发能力（核心线程数2，最大线程数5），不会阻塞
        
        return taskId;
    }
    
    /**
     * 获取需要同步的日期列表（公共方法）
     * 1. 查询已存在的数据
     * 2. 判断哪些是完整数据，哪些是非完整数据
     * 3. 删除非完整数据
     * 4. 获取非周末日期列表并过滤出需要同步的日期
     * 
     * @param dto 同步参数
     * @return 需要同步的日期列表
     */
    private List<String> getSyncDateList(SyncSecurityDayMinuteTradeDto dto) {
        // 已经存在的完整数据列表，完整数据是指包含所有分钟数据的记录，共241条
        List<String> existFullDataList = Lists.newArrayList();
        // 已经存在的非完整数据列表
        List<String> existNonFullDataList = Lists.newArrayList();

        // 查询已存在的数据
        LambdaQueryWrapper<SecurityDayMinuteTrade> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(SecurityDayMinuteTrade::getSymbol, dto.getSymbol());
        queryWrapper.ge(SecurityDayMinuteTrade::getDate, dto.getStartDate());
        queryWrapper.le(SecurityDayMinuteTrade::getDate, dto.getEndDate());
        List<SecurityDayMinuteTrade> minuteTradeList = securityDayMinuteTradeMapper.selectList(queryWrapper);
        
        // 按日期分组，判断数据完整性
        Map<Date, List<SecurityDayMinuteTrade>> groupByDate = minuteTradeList.stream()
                .collect(Collectors.groupingBy(SecurityDayMinuteTrade::getDate));
        for (Map.Entry<Date, List<SecurityDayMinuteTrade>> entry : groupByDate.entrySet()) {
            Date date = entry.getKey();
            int count = entry.getValue().size();
            if (ApiConstant.SECURITY_MINUTE_TRADE_COUNT == count) {
                existFullDataList.add(DateUtil.parseDateToStr(date));
            } else {
                log.info("证券【{} {}】分钟数据不完整，需要重新爬取", dto.getSymbol(), DateUtil.parseDateToStr(date));
                existNonFullDataList.add(DateUtil.parseDateToStr(date));
            }
        }
        
        // 删除已经存在的非完整数据
        if (!CollectionUtils.isEmpty(existNonFullDataList)) {
            LambdaQueryWrapper<SecurityDayMinuteTrade> deleteQuery = new LambdaQueryWrapper<>();
            deleteQuery.eq(SecurityDayMinuteTrade::getSymbol, dto.getSymbol());
            deleteQuery.in(SecurityDayMinuteTrade::getDate, existNonFullDataList);
            securityDayMinuteTradeMapper.delete(deleteQuery);
        }

        // 通过交易日历获取交易日期列表，并过滤出需要同步的日期（排除已存在的完整数据）
        Date startDate = DateUtil.parseDate(dto.getStartDate());
        Date endDate = DateUtil.parseDate(dto.getEndDate());
        List<SecurityTradingCalendar> tradingDays = securityTradingCalendarService.selectTradingDays(startDate, endDate);
        List<String> tradingDateList = tradingDays.stream()
                .map(calendar -> DateUtil.parseDateToStr(calendar.getDate()))
                .collect(Collectors.toList());
        return tradingDateList.stream()
                .filter(e -> !existFullDataList.contains(e))
                .collect(Collectors.toList());
    }
    
    /**
     * 使用多线程并发同步分时数据
     * @param dto 同步参数
     * @param syncDateList 需要同步的日期列表
     * @param taskId 任务ID（可选，如果为null则不更新任务进度）
     */
    private void syncListWithMultiThread(SyncSecurityDayMinuteTradeDto dto, List<String> syncDateList, String taskId) {
        // 将日期列表分批，每批处理一个日期（或者可以每批处理多个日期）
        List<Future<List<SecurityDayMinuteTrade>>> futures = new ArrayList<>();
        
        for (String date : syncDateList) {
            Future<List<SecurityDayMinuteTrade>> future = executorService.submit(() -> {
                try {
                    log.info("开始同步分时数据，证券代码：{}，日期：{}", dto.getSymbol(), date);
                    
                    // 更新任务进度
                    if (taskId != null) {
                        int currentSynced = syncTaskManager.getTask(taskId).getSyncedDays().get();
                        syncTaskManager.updateProgress(taskId, currentSynced, date);
                    }
                    
                    // 调用爬虫服务获取单个日期的数据
                    List<String> dateList = Lists.newArrayList(date);
                    List<SecurityDayMinuteTradeResponse> responseList = crawService.crawList(dto.getSymbol(), dateList);
                    
                    if (CollectionUtils.isEmpty(responseList)) {
                        log.warn("证券【{} {}】未获取到数据", dto.getSymbol(), date);
                        if (taskId != null) {
                            // 未获取到数据，但日期已处理，计入已处理天数
                            syncTaskManager.incrementProcessedDays(taskId);
                        }
                        return Lists.newArrayList();
                    }
                    
                    // 转换为实体对象
                    List<SecurityDayMinuteTrade> insertList = SecurityDayMinuteTradeConvert.INSTANCE.toSecurityDayMinuteTradeList(responseList);
                    
                    // 设置字段
                    Date now = new Date();
                    for (SecurityDayMinuteTrade trade : insertList) {
                        trade.setSymbol(dto.getSymbol());
                        trade.setCode(dto.getSymbol().substring(2));
                        trade.setCreateTime(now);
                        trade.setUpdateTime(now);
                    }
                    
                    // 批量插入数据
                    if (!CollectionUtils.isEmpty(insertList)) {
                        for (SecurityDayMinuteTrade trade : insertList) {
                            securityDayMinuteTradeMapper.insert(trade);
                        }
                        log.info("同步分时数据成功，证券代码：{}，日期：{}，数据条数：{}", 
                                dto.getSymbol(), date, insertList.size());
                        
                        // 更新任务进度（成功同步）
                        if (taskId != null) {
                            syncTaskManager.incrementSyncedDays(taskId);
                            syncTaskManager.incrementProcessedDays(taskId);
                        }
                    } else {
                        // insertList为空，但已经处理过这个日期了
                        if (taskId != null) {
                            syncTaskManager.incrementProcessedDays(taskId);
                        }
                    }
                    
                    return insertList;
                } catch (Exception e) {
                    log.error("同步分时数据失败，证券代码：{}，日期：{}", dto.getSymbol(), date, e);
                    if (taskId != null) {
                        // 同步失败，但日期已处理，计入失败和已处理
                        syncTaskManager.incrementFailedDays(taskId);
                        syncTaskManager.incrementProcessedDays(taskId);
                    }
                    throw new RuntimeException("同步分时数据失败：" + e.getMessage(), e);
                }
            });
            
            futures.add(future);
        }
        
        // 等待所有任务完成
        for (Future<List<SecurityDayMinuteTrade>> future : futures) {
            try {
                future.get(); // 等待任务完成，如果失败会抛出异常
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.error("同步分时数据被中断", e);
                if (taskId != null) {
                    syncTaskManager.markTaskFailed(taskId, "同步分时数据被中断");
                }
                throw new RuntimeException("同步分时数据被中断", e);
            } catch (ExecutionException e) {
                log.error("同步分时数据执行失败", e);
                if (taskId != null) {
                    // 同步失败，但日期已处理
                    syncTaskManager.incrementFailedDays(taskId);
                    syncTaskManager.incrementProcessedDays(taskId);
                    // 不标记整个任务失败，继续处理其他日期
                }
                // 不抛出异常，继续处理其他日期
            }
        }
        
        // 所有任务完成后，再次检查并更新任务状态（确保状态正确）
        if (taskId != null) {
            SyncTaskManager.TaskInfo taskInfo = syncTaskManager.getTask(taskId);
            if (taskInfo != null && "RUNNING".equals(taskInfo.getStatus())) {
                int processedDays = taskInfo.getProcessedDays().get();
                if (processedDays >= taskInfo.getTotalDays()) {
                    // 所有日期都已处理完成
                    taskInfo.setStatus("COMPLETED");
                    taskInfo.setEndTime(System.currentTimeMillis());
                    log.info("同步任务完成，任务ID：{}，成功：{}，失败：{}，总处理：{}", 
                            taskId, taskInfo.getSyncedDays().get(), taskInfo.getFailedDays().get(), processedDays);
                }
            }
        }
        
        log.info("所有分时数据同步完成，证券代码：{}，共同步{}个日期", dto.getSymbol(), syncDateList.size());
    }

    /**
     * 设置查询条件
     */
    private LambdaQueryWrapper<SecurityDayMinuteTrade> buildQueryConditions( SecurityDayMinuteTradeDto dto) {
        LambdaQueryWrapper<SecurityDayMinuteTrade> queryWrapper = new LambdaQueryWrapper<>();
        if (StringUtils.isNotBlank(dto.getSymbol())) {
            queryWrapper.eq(SecurityDayMinuteTrade::getSymbol, dto.getSymbol());
        }
        if (StringUtils.isNotBlank(dto.getCode())) {
            queryWrapper.eq(SecurityDayMinuteTrade::getCode, dto.getCode());
        }
        if (StringUtils.isNotBlank(dto.getSecurityName())) {
            queryWrapper.like(SecurityDayMinuteTrade::getName, dto.getSecurityName());
        }
        if (StringUtils.isNotBlank(dto.getStratDate())) {
            queryWrapper.ge(SecurityDayMinuteTrade::getDate, DateUtils.dateTime(DateUtils.YYYY_MM_DD, dto.getStratDate()));
        }
        if (StringUtils.isNotBlank(dto.getEndDate())) {
            queryWrapper.le(SecurityDayMinuteTrade::getDate, DateUtils.dateTime(DateUtils.YYYY_MM_DD, dto.getEndDate()));
        }
        return queryWrapper;
    }

    @Override
    public List<String> getInvalidTradeDates(String symbol, List<SecurityTradingCalendar> tradingDays, 
                                               List<SecurityDayMinuteTrade> tradeList) {
        // 按日期分组
        Map<Date, List<SecurityDayMinuteTrade>> tradeByDate = tradeList.stream()
                .collect(Collectors.groupingBy(SecurityDayMinuteTrade::getDate));
        
        List<String> invalidDates = new ArrayList<>();
        
        // 检查每个交易日是否有数据
        for (SecurityTradingCalendar tradingDay : tradingDays) {
            Date date = tradingDay.getDate();
            String dateStr = DateUtil.parseDateToStr(date);
            List<SecurityDayMinuteTrade> dayTradeList = tradeByDate.get(date);
            
            if (dayTradeList == null || dayTradeList.isEmpty()) {
                // 该交易日没有数据
                invalidDates.add(dateStr);
            } else {
                // 检查数据是否完整（应该是241条）
                int count = dayTradeList.size();
                if (count < ApiConstant.SECURITY_MINUTE_TRADE_COUNT) {
                    // 数据不完整，也加入列表
                    invalidDates.add(dateStr);
                }
            }
        }
        
        return invalidDates;
    }

}
